package com.lintf.distributed.transaction.inventory.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.lintf.distributed.transaction.inventory.entity.DeadLetterMessage;
import com.lintf.distributed.transaction.inventory.enums.DeadLetterStatus;
import com.lintf.distributed.transaction.inventory.exception.BusinessException;
import com.lintf.distributed.transaction.inventory.mapper.DeadLetterMessageMapper;
import com.lintf.distributed.transaction.inventory.service.DeadLetterService;
import lombok.RequiredArgsConstructor;
import net.sf.jsqlparser.util.validation.metadata.DatabaseException;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.Objects;
import java.util.concurrent.TimeoutException;

/**
 * @ author lintf
 * @ system Mac
 * @ data 2025/6/26
 * @ package_name com.lintf.distributed.transaction.inventory.service.impl
 * @ project_name distributed-transaction-demo
 * @ desc
 */
@Service
@RequiredArgsConstructor
public class DeadLetterServiceImpl extends ServiceImpl<DeadLetterMessageMapper, DeadLetterMessage> implements DeadLetterService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Transactional
    @Override
    public DeadLetterMessage saveDeadLetter(String key, String value, String topic, Throwable cause, Integer retryCount) {
        DeadLetterMessage dlMessage = new DeadLetterMessage();
        dlMessage.setMessageKey(key);
        dlMessage.setOriginalTopic(topic);
        dlMessage.setMessageBody(value);
        dlMessage.setFailureTime(new Date());
        dlMessage.setFailureReason(getFailureReason(cause));
        dlMessage.setExceptionStack(ExceptionUtils.getStackTrace(cause));
        dlMessage.setStatus(DeadLetterStatus.NEW);
        dlMessage.setRetryCount(retryCount);
        dlMessage.setRetryMax(5); // 允许最多5次人工重试
        dlMessage.setCreatedAt(new Date());
        dlMessage.setUpdatedAt(new Date());
        save(dlMessage);
        return dlMessage;
    }

    private String getFailureReason(Throwable cause) {
        if (cause == null) return "Unknown error";

        if (cause instanceof BusinessException) {
            return "业务异常: " + cause.getMessage();
        } else if (cause instanceof TimeoutException) {
            return "超时异常: " + cause.getMessage();
        } else if (cause instanceof DatabaseException) {
            return "数据库异常: " + cause.getMessage();
        }
        return "系统异常: " + cause.getMessage();
    }

    // 人工重试接口
    @Transactional
    @Override
    public void retryMessage(Long deadLetterId) {
        DeadLetterMessage dlMessage = getById(deadLetterId);
        if (Objects.isNull(dlMessage)) {
            throw new BusinessException("死信消息不存在");
        }

        if (dlMessage.getRetryCount() >= dlMessage.getRetryMax()) {
            throw new BusinessException("超过最大重试次数");
        }

        try {
            // 发送回原队列重试
            kafkaTemplate.send(
                    dlMessage.getOriginalTopic(),
                    dlMessage.getMessageKey(),
                    dlMessage.getMessageBody()
            );

            // 更新重试状态
            dlMessage.setRetryCount(dlMessage.getRetryCount() + 1);
            dlMessage.setLastRetryTime(new Date());
            updateById(dlMessage);

        } catch (Exception e) {
            log.error("死信消息重试发送失败", e);
            throw new BusinessException("重试失败: " + e.getMessage());
        }
    }
}
